package io.kiki.sba.registry.server.data_store.multi.cluster.slot;

import com.google.common.collect.Sets;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.console.MultiSegmentSyncSwitch;
import io.kiki.sba.registry.common.model.constants.MultiValueConstants;
import io.kiki.sba.registry.common.model.metaserver.MultiClusterSyncInfo;
import io.kiki.sba.registry.common.model.multi.cluster.RemoteSlotsStatus;
import io.kiki.sba.registry.common.model.slot.*;
import io.kiki.sba.registry.common.model.slot.filter.*;
import io.kiki.sba.registry.common.model.slot.func.SlotFunction;
import io.kiki.sba.registry.common.model.slot.func.SlotFunctionRegistry;
import io.kiki.sba.registry.exception.UnSupportOperationException;
import io.kiki.sba.registry.server.data_store.bootstrap.MultiClusterDataServerConfig;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.change.DataChangeEventCenter;
import io.kiki.sba.registry.server.data_store.multi.cluster.exchanger.RemoteDataNodeExchanger;
import io.kiki.sba.registry.server.data_store.multi.cluster.executor.MultiClusterExecutorManager;
import io.kiki.sba.registry.server.data_store.multi.cluster.slot.MultiClusterSlotMetrics.SyncType;
import io.kiki.sba.registry.server.data_store.multi.cluster.sync.info.FetchMultiSyncService;
import io.kiki.sba.registry.server.data_store.slot.*;
import io.kiki.sba.registry.server.data_store.slot.SlotManagerImpl.ISlotState;
import io.kiki.sba.registry.server.shared.remoting.ClientSideExchanger;
import io.kiki.sba.registry.store.api.meta.MultiClusterSyncRepository;
import io.kiki.sba.registry.task.KeyedTask;
import io.kiki.sba.registry.util.*;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;


public class MultiClusterSlotManagerImpl implements MultiClusterSlotManager {

    private static final Logger logger = LoggerFactory.getLogger(MultiClusterSlotManagerImpl.class);

    private static final int MAX_DATAINFOID_SYNCING_FAIL_RETRY_SIZE = 1000;


    private final Map<String, RemoteSlotsStorage> dataCenterToRemoteSlotsStorageMap = new ConcurrentHashMap<>();
    private final SlotFunction slotFunction = SlotFunctionRegistry.getSlotFunction();
    @Getter
    private final RemoteSyncingWatchDog remoteSyncingWatchDog = new RemoteSyncingWatchDog();
    @Autowired
    private ServerConfig serverConfig;
    @Autowired
    private MultiClusterDataServerConfig multiClusterDataServerConfig;
    @Autowired
    private DataChangeEventCenter dataChangeEventCenter;
    @Resource
    private DatumStorageDelegate datumStorageDelegate;
    @Autowired
    private SlotChangeListenerManager slotChangeListenerManager;
    @Autowired
    private RemoteDataNodeExchanger remoteDataNodeExchanger;
    @Autowired
    private MultiClusterExecutorManager multiClusterExecutorManager;
    @Autowired
    private MultiSyncDataAcceptorManager multiSyncDataAcceptorManager;
    @Autowired
    private FetchMultiSyncService fetchMultiSyncService;
    @Autowired
    private MultiClusterSyncRepository multiClusterSyncRepository;
    @Autowired
    private SlotManager slotManager;

    @PostConstruct
    public void init() {
        ConcurrentUtils.createDaemonThread("RemoteSyncingWatchDog", remoteSyncingWatchDog).start();
    }

    @Override
    public int slotOf(String dataInfoId) {
        return slotFunction.slotOf(dataInfoId);
    }

    @Override
    public Slot getSlot(String dataCenter, int slotId) {
        RemoteSlotsStorage remoteSlotsStorage = dataCenterToRemoteSlotsStorageMap.get(dataCenter);
        if (remoteSlotsStorage == null) {
            return null;
        }

        final RemoteSlotsStates remoteSlotsStates = remoteSlotsStorage.SlotsStates;
        if (remoteSlotsStates == null) {
            return null;
        }
        RemoteSlotStates remoteSlotStates = remoteSlotsStates.slotStates.get(slotId);
        return remoteSlotStates == null ? null : remoteSlotStates.slot;
    }

    @Override
    public SlotAccess checkSlotAccess(String dataCenter, int slotId, long srcSlotEpoch, long srcLeaderEpoch) {
        RemoteSlotsStorage remoteSlotsStorage = dataCenterToRemoteSlotsStorageMap.get(dataCenter);
        if (remoteSlotsStorage == null) {
            return null;
        }

        final RemoteSlotsStates remoteSlotsStates = remoteSlotsStorage.SlotsStates;
        if (remoteSlotsStates == null) {
            return new SlotAccess(slotId, Slots.INIT.getEpoch(), SlotAccess.Status.unsupport, -1);
        }
        Tuple<Slots, RemoteSlotStates> tuple = remoteSlotsStates.get(slotId);

        return checkSlotAccess(slotId, tuple.o1.getEpoch(), tuple.o2, srcLeaderEpoch);
    }

    public SlotAccess checkSlotAccess(int slotId, long currentslotsEpoch, ISlotState slotState, long srcLeaderEpoch) {
        if (slotState == null) {
            return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.moved, -1);
        }
        final Slot slot = slotState.getSlot();
        if (!localIsLeader(slot)) {
            return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.moved, slot.getLeaderEpoch());
        }
        if (!slotState.isMigrated()) {
            return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.migrating, slot.getLeaderEpoch());
        }
        if (slot.getLeaderEpoch() != srcLeaderEpoch) {
            return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.mismatch, slot.getLeaderEpoch());
        }
        return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.accept, slot.getLeaderEpoch());
    }

    private boolean localIsLeader(Slot slot) {
        return slotManager.isLeader(serverConfig.getDataCenter(), slot.getId());
    }

    @Override
    public boolean isLeader(String dataCenter, int slotId) {
        RemoteSlotsStorage remoteSlotsStorage = dataCenterToRemoteSlotsStorageMap.get(dataCenter);
        if (remoteSlotsStorage == null) {
            return false;
        }

        final RemoteSlotsStates remoteSlotsStates = remoteSlotsStorage.SlotsStates;
        if (remoteSlotsStates == null) {
            return false;
        }
        RemoteSlotStates remoteSlotStates = remoteSlotsStates.get(slotId).o2;
        return remoteSlotStates != null && localIsLeader(remoteSlotStates.slot);
    }

    @Override
    public boolean isFollower(String dataCenter, int slotId) {
        throw new UnSupportOperationException(StringFormatter.format("MultiClusterSlotManagerImpl.isFollower {}/{}", dataCenter, slotId));
    }

    @Override
    public Tuple<Long, List<BaseSlotStatus>> getslotsEpochAndStatuses(String dataCenter) {
        RemoteSlotsStorage remoteSlotsStorage = dataCenterToRemoteSlotsStorageMap.get(dataCenter);
        if (remoteSlotsStorage == null) {
            return new Tuple<>(Slots.INIT.getEpoch(), Collections.emptyList());
        }

        long slotsEpoch = remoteSlotsStorage.SlotsStates.getSlotsEpoch();
        List<BaseSlotStatus> slotStatusList = remoteSlotsStorage.SlotsStates.getSlotStatuses();
        return new Tuple<>(slotsEpoch, slotStatusList);
    }

    /**
     * get remote cluster Slots epoch
     */
    @Override
    public Map<String, Long> getSlotsEpoch() {
        Map<String, Long> slotsEpochMap = new HashMap<>(dataCenterToRemoteSlotsStorageMap.size());
        for (Entry<String, RemoteSlotsStorage> entry : dataCenterToRemoteSlotsStorageMap.entrySet()) {

            long epoch;
            if (entry.getValue() == null || entry.getValue().SlotsStates == null) {
                epoch = Slots.INIT.getEpoch();
            } else {
                epoch = entry.getValue().SlotsStates.slots.getEpoch();
            }
            slotsEpochMap.put(entry.getKey(), epoch);
        }
        return slotsEpochMap;
    }

    /**
     * 1.add new dataCenter slot table to remoteSlotsStates 2.update exist dataCenter slot table
     * 3.important: remove slot table which not exist in meta
     *
     * @param map remoteSlotsStatus
     */
    @Override
    public void updateSlots(Map<String, RemoteSlotsStatus> map) {

        Set<String> tobeRemove = Sets.difference(dataCenterToRemoteSlotsStorageMap.keySet(), map.keySet());

        boolean wakeup = false;
        for (Entry<String, RemoteSlotsStatus> entry : map.entrySet()) {
            RemoteSlotsStatus remoteSlotsStatus = entry.getValue();
            if (remoteSlotsStatus.isSlotsEpochConflict()) {
                // local.slotsEpoch > meta.slotsEpoch
                // it should not happen, print error log and restart data server;
                logger.error("[updateSlots]meta remote slot table status conflict: {}", remoteSlotsStatus);
                continue;
            }

            String dataCenter = entry.getKey();

            RemoteSlotsStorage remoteSlotsStorage = dataCenterToRemoteSlotsStorageMap.computeIfAbsent(dataCenter, k -> new RemoteSlotsStorage(new RemoteSlotsStates(dataCenter), new AtomicReference<>()));
            Slots slots = remoteSlotsStorage.SlotsStates.slots;
            // upgrade=false, slot table not change
            // upgrade=true, but data had accept a bigger version than return value
            if (!remoteSlotsStatus.isSlotsUpgrade() || slots.getEpoch() >= remoteSlotsStatus.getSlotsEpoch()) {
                continue;
            }

            // check updating slot table
            AtomicReference<Slots> updating = remoteSlotsStorage.updating;
            Slots updatingSlots = updating.get();
            if (updatingSlots != null && updatingSlots.getEpoch() >= remoteSlotsStatus.getSlotsEpoch()) {
                continue;
            }
            Slots toBeUpdate = remoteSlotsStatus.getSlots();

            if (!checkSlot(slots, updatingSlots, toBeUpdate)) {
                continue;
            }

            if (updating.compareAndSet(updatingSlots, toBeUpdate)) {
                wakeup = true;
                logger.info("[updateSlots]updating remote slot table, dataCenter={}, prev={}, new={}", dataCenter, slots, toBeUpdate);
            }
        }
        if (wakeup) {
            remoteSyncingWatchDog.wakeup();
        }

        processRemove(tobeRemove);
    }

    private void processRemove(Set<String> tobeRemove) {
        if (CollectionUtils.isEmpty(tobeRemove)) {
            return;
        }
        Set<MultiClusterSyncInfo> syncInfos = multiClusterSyncRepository.queryLocalMultiClusterSyncInfoSet();
        Set<String> syncing = syncInfos.stream().map(MultiClusterSyncInfo::getRemoteDataCenter).collect(Collectors.toSet());
        for (String remove : tobeRemove) {
            if (syncing.contains(remove)) {
                logger.error("dataCenter:{} remove is forbidden.", remove);
                continue;
            }

            logger.info("start to remove dataCenter:{} datum and Slots.", remove);
            boolean removeSuccess = datumStorageDelegate.removeStorage(remove);
            if (!removeSuccess) {
                logger.error("dataCenter:{} remove storage fail.", remove);
                continue;
            }
            dataCenterToRemoteSlotsStorageMap.remove(remove);
            logger.info("remove dataCenter:{} datum and Slots success.", remove);
        }
    }

    @Override
    public void dataChangeNotify(String dataCenter, Set<String> dataInfoIds) {
        if (!fetchMultiSyncService.multiSync(dataCenter)) {
            logger.info("[syncDisable]dataCenter: {} data change:{} notify", dataCenter, dataInfoIds);
            return;
        }

        Map<Integer, Set<String>> slotDataInfoIds = new HashMap<>();
        for (String dataInfoId : dataInfoIds) {
            SyncSlotAcceptorManager syncSlotAcceptorManager = multiSyncDataAcceptorManager.getSyncSlotAcceptorManager(dataCenter);
            if (syncSlotAcceptorManager == null || !syncSlotAcceptorManager.accept(SyncAcceptorRequest.buildRequest(dataInfoId))) {
                logger.info("[NotAccept]dataCenter: {} data change:{} notify", dataCenter, dataInfoId);
                continue;
            }

            int slotId = slotOf(dataInfoId);
            Set<String> set = slotDataInfoIds.computeIfAbsent(slotId, k -> new HashSet<>());
            set.add(dataInfoId);
        }

        RemoteSlotsStorage remoteSlotsStorage = dataCenterToRemoteSlotsStorageMap.get(dataCenter);
        if (remoteSlotsStorage == null) {
            logger.info("[skip]dataCenter: {} data change:{} notify", dataCenter, dataInfoIds);
            return;
        }

        for (Entry<Integer, Set<String>> entry : slotDataInfoIds.entrySet()) {
            RemoteSlotStates remoteSlotStates = remoteSlotsStorage.SlotsStates.getSlotStates(entry.getKey());
            if (remoteSlotStates == null) {
                logger.info("[skip]dataCenter: {}, slotId:{},  data change:{} notify", dataCenter, entry.getKey(), dataInfoIds);
                continue;
            }
            remoteSlotStates.addPending(entry.getValue());
            logger.info("dataCenter: {}, slotId:{},  data change:{} add to pending.", dataCenter, entry.getKey(), dataInfoIds);
        }
    }

    private boolean checkSlot(Slots cur, Slots updating, Slots update) {
        try {
            cur.assertSlotLessThan(update);
            if (updating != null) {
                update.assertSlotLessThan(update);
            }
            return true;
        } catch (RuntimeException e) {
            logger.error("[checkSlot]assert slot fail, cur: {}, updating: {}, update: {}", cur, updating, update, e);
            return false;
        }
    }

    /**
     * update remote state
     */
    boolean doUpdating() {
        for (Entry<String, RemoteSlotsStorage> entry : dataCenterToRemoteSlotsStorageMap.entrySet()) {
            String dataCenter = entry.getKey();
            Slots update = entry.getValue().updating.getAndSet(null);
            if (update == null) {
                continue;
            }
            RemoteSlotsStates remoteSlotsStates = entry.getValue().SlotsStates;
            Slots current = remoteSlotsStates.slots;
            if (update.getEpoch() <= current.getEpoch()) {
                logger.warn("skip remoteDataCenter={}, updating={}, current={}", dataCenter, update.getEpoch(), current.getEpoch());
                continue;
            }
            remoteSlotsStates.updateSlotState(update);
        }
        return true;
    }

    // compare with slot manager leader slotId
    private void watchLocalSegSlot() {
        Set<Integer> leaderIds = slotManager.leaderSlotIds();
        for (Entry<String, RemoteSlotsStorage> entry : dataCenterToRemoteSlotsStorageMap.entrySet()) {
            RemoteSlotsStates remoteSlotsStates = entry.getValue().SlotsStates;
            Set<Integer> currents = remoteSlotsStates.slotIds();
            Set<Integer> adds = new HashSet<>(Sets.difference(leaderIds, currents));
            Set<Integer> removes = new HashSet<>(Sets.difference(currents, leaderIds));

            if (!CollectionUtils.isEmpty(adds) || !CollectionUtils.isEmpty(removes)) {
                remoteSlotsStates.addSlotState(adds);
                remoteSlotsStates.removeSlotState(removes);
                logger.info("[doSlotStatesUpdate]dataCenter={}, adds={}, removes={}", entry.getKey(), adds, removes);
            }
        }
    }

    void doSyncRemoteLeader() {
        final int remoteSyncLeaderMs = multiClusterDataServerConfig.getSyncRemoteSlotLeaderIntervalSecs() * 1000;

        for (Entry<String, RemoteSlotsStorage> entry : dataCenterToRemoteSlotsStorageMap.entrySet()) {
            String remoteDataCenter = entry.getKey();
            if (!fetchMultiSyncService.multiSync(remoteDataCenter)) {
                continue;
            }

            RemoteSlotsStates remoteSlotsStates = entry.getValue().SlotsStates;
            for (RemoteSlotStates remoteSlotStates : remoteSlotsStates.slotStates.values()) {
                try {
                    syncRemoteDataIds(remoteDataCenter, remoteSlotStates, remoteSlotsStates.slots.getEpoch());
                    syncRemote(remoteDataCenter, remoteSlotStates, remoteSyncLeaderMs, remoteSlotsStates.slots.getEpoch());
                } catch (Throwable t) {
                    logger.error("[syncRemoteLeader]remoteDataCenter={}, slotId={} sync error.", entry.getKey(), remoteSlotStates.slotId, t);
                }
            }
        }
    }

    private Set<String> getTobeSyncs(RemoteSlotStates remoteSlotStates, KeyedTask<SyncDataIdTask> syncDataIdTask) {
        if (syncDataIdTask != null && syncDataIdTask.isFailed()) {

            Set<String> syncedFail = syncDataIdTask.getRunnable().syncing;
            if (syncedFail.size() < MAX_DATAINFOID_SYNCING_FAIL_RETRY_SIZE) {
                logger.error("getTobeSyncs repending syncDataIdTask:{} to retry", syncedFail);
                remoteSlotStates.addPending(syncedFail);
            } else {
                logger.error("getTobeSyncs syncDataIdTask.size={} > {}", syncedFail.size(), MAX_DATAINFOID_SYNCING_FAIL_RETRY_SIZE);
            }
        }
        return remoteSlotStates.pendingDataInfoIds.getAndReset();
    }

    private Set<SyncSlotAcceptor> acceptors(String remoteDataCenter, Set<String> syncing) {
        HashSet<SyncSlotAcceptor> hashSet = new HashSet<>(Arrays.asList(MultiValueConstants.DATUM_SYNCER_SOURCE_FILTER));
        MultiSegmentSyncSwitch multiSegmentSyncSwitch = fetchMultiSyncService.getMultiSyncSwitch(remoteDataCenter);

        hashSet.add(new SyncSlotDataInfoIdAcceptor(syncing, multiSegmentSyncSwitch.getIgnoreDataInfoIds()));

        return hashSet;
    }

    private void syncRemoteDataIds(String remoteDataCenter, RemoteSlotStates remoteSlotStates, long slotsEpoch) {
        final Slot slot = remoteSlotStates.slot;
        final KeyedTask<SyncDataIdTask> syncDataIdTask = remoteSlotStates.syncDataIdTask;

        if (syncDataIdTask != null && syncDataIdTask.isFinished()) {
            remoteSlotStates.completeSyncRemoteDataIdTask();
        }
        // 1.syncDataIdTask == null, state.pending is not empty, execute new task;
        // 2.syncDataIdTask.isFinished(), state.pending is not empty, execute new task(200ms to
        //   execute a new task);
        // 3.syncDataIdTask.isNotFinished(), waiting task to finish
        if (syncDataIdTask == null || syncDataIdTask.isFinished()) {

            Set<String> tobeSyncs = getTobeSyncs(remoteSlotStates, syncDataIdTask);
            if (CollectionUtils.isEmpty(tobeSyncs)) {
                return;
            }
            logger.info("tobeSyncs dataInfoIds:{}", tobeSyncs);
            SlotDiffSyncer slotDiffSyncer = new SlotDiffSyncer(serverConfig, datumStorageDelegate, dataChangeEventCenter, null, multiSyncDataAcceptorManager
                    .new RemoteSyncDataAcceptorManager(acceptors(remoteDataCenter, tobeSyncs)));
            SyncContinues continues = () -> localIsLeader(slot);
            SyncDataIdTask task = new SyncDataIdTask(serverConfig.getDataCenter(), remoteDataCenter, slotsEpoch, slot, slotDiffSyncer, tobeSyncs, remoteDataNodeExchanger, continues);
            remoteSlotStates.syncDataIdTask = multiClusterExecutorManager.getRemoteSyncDataIdExecutor().execute(slot.getId(), task);
            MultiClusterSlotMetrics.syncAccess(remoteDataCenter, SyncType.sync_delta);
            return;
        }

        if (System.currentTimeMillis() - syncDataIdTask.getCreateTime() > 1500) {
            // the sync leader is running more than 1500ms, print
            logger.info("remoteDataCenter={}, slotId={}, dataIds={}, sync-dataid running, {}", remoteDataCenter, slot.getId(), syncDataIdTask.getRunnable().syncing, syncDataIdTask);
        }
    }

    void syncRemote(String remoteDataCenter, RemoteSlotStates remoteSlotStates, int remoteSyncLeaderMs, long slotsEpoch) {
        final Slot slot = remoteSlotStates.slot;
        final KeyedTask<SyncLeaderTask> syncRemoteTask = remoteSlotStates.syncRemoteTask;

        if (syncRemoteTask == null || syncRemoteTask.isOverAfter(remoteSyncLeaderMs)) {
            SlotDiffSyncer syncer = new SlotDiffSyncer(serverConfig, datumStorageDelegate, dataChangeEventCenter, null, multiSyncDataAcceptorManager.getSyncSlotAcceptorManager(remoteDataCenter));
            SyncContinues continues = () -> localIsLeader(slot);
            SyncLeaderTask task = new SyncLeaderTask(serverConfig.getDataCenter(), remoteDataCenter, slotsEpoch, slot, syncer, remoteDataNodeExchanger, continues);
            remoteSlotStates.syncRemoteTask = multiClusterExecutorManager.getRemoteSyncLeaderExecutor().execute(slot.getId(), task);
            MultiClusterSlotMetrics.syncAccess(remoteDataCenter, SyncType.sync_all);
            return;
        }

        if (syncRemoteTask.isFinished()) {
            remoteSlotStates.completeSyncRemoteLeaderTask();
        } else {
            if (System.currentTimeMillis() - syncRemoteTask.getCreateTime() > 5000) {
                // the sync leader is running more than 5secs, print
                logger.info("remoteDataCenter={}, slotId={}, sync-leader running, {}", remoteDataCenter, slot.getId(), syncRemoteTask);
            }
        }
    }


    static final class RemoteSlotStates implements ISlotState {
        final String remoteDataCenter;
        final int slotId;
        volatile Slot slot;
        volatile boolean synced;
        volatile long lastSuccessSyncRemoteTime = -1L;
        volatile KeyedTask<SyncLeaderTask> syncRemoteTask;

        AtomicSet<String> pendingDataInfoIds = new AtomicSet<>();
        volatile long lastSuccessDataInfoIdTime = -1L;
        volatile KeyedTask<SyncDataIdTask> syncDataIdTask;

        RemoteSlotStates(String remoteDataCenter, Slot slot) {
            this.remoteDataCenter = remoteDataCenter;
            this.slotId = slot.getId();
            this.slot = slot;
        }

        void update(Slot update) {
            StringUtil.checkEquals(slotId, update.getId(), "slot.id");
            if (slot.getLeaderEpoch() != update.getLeaderEpoch()) {
                this.synced = false;
                this.lastSuccessSyncRemoteTime = -1L;
                this.syncRemoteTask = null;

                // clear task when slot update;
                // the SyncLeaderTask will sync all;
                this.pendingDataInfoIds.getAndReset();
                this.lastSuccessDataInfoIdTime = -1L;
                this.syncDataIdTask = null;
            }
            this.slot = update;
            logger.info("remoteDataCenter={}, update slot={}", remoteDataCenter, update);
        }

        void completeSyncRemoteLeaderTask() {
            if (syncRemoteTask != null && syncRemoteTask.isSuccess()) {
                this.lastSuccessSyncRemoteTime = syncRemoteTask.getEndTime();
                this.synced = true;
            }
        }

        void completeSyncRemoteDataIdTask() {
            if (syncDataIdTask == null || !syncDataIdTask.isFinished()) {
                return;
            }

            if (syncDataIdTask.isSuccess()) {
                this.lastSuccessDataInfoIdTime = syncDataIdTask.getEndTime();
            }
        }

        public synchronized void addPending(String dataInfoId) {
            pendingDataInfoIds.add(dataInfoId);
        }

        public synchronized void addPending(Set<String> dataInfoIds) {
            pendingDataInfoIds.addAll(dataInfoIds);
        }

        public synchronized boolean hasPending() {
            return pendingDataInfoIds.size() > 0;
        }

        @Override
        public boolean isMigrated() {
            return this.synced;
        }

        @Override
        public Slot getSlot() {
            return this.slot;
        }
    }

    final class RemoteSlotsStorage {
        @Getter
        private final RemoteSlotsStates SlotsStates;

        private final AtomicReference<Slots> updating;

        public RemoteSlotsStorage(RemoteSlotsStates SlotsStates, AtomicReference<Slots> updating) {
            this.SlotsStates = SlotsStates;
            this.updating = updating;
        }


    }

    final class RemoteSlotsStates {
        final String dataCenter;
        // map<slotId, RemoteSlotStates>, slot belong to me
        final Map<Integer, RemoteSlotStates> slotStates = new ConcurrentHashMap<>();
        private final ReentrantReadWriteLock updateLock = new ReentrantReadWriteLock();
        private final ReentrantReadWriteLock.WriteLock writeLock = updateLock.writeLock();
        private final ReentrantReadWriteLock.ReadLock readLock = updateLock.readLock();
        // save all slot
        volatile Slots slots = Slots.INIT;

        public RemoteSlotsStates(String dataCenter) {
            this.dataCenter = dataCenter;
        }

        boolean updateSlotState(Slots update) {

            // filter slot belong to me
            Slots mySlots = update.filter(slotManager.leaderSlotIds());
            writeLock.lock();
            try {
                for (Slot slot : mySlots.getSlotIdToSlotMap()) {
                    listenAdd(dataCenter, slot.getId());
                    RemoteSlotStates state = slotStates.computeIfAbsent(slot.getId(), k -> {
                        logger.info("[updateSlotState]add dataCenter={}, slot={}", dataCenter, slot);
                        return new RemoteSlotStates(dataCenter, slot);
                    });
                    state.update(slot);
                }

                this.slots = update;
                MultiClusterSlotMetrics.observeRemoteLeaderAssignGauge(dataCenter, this.slots.getSlotNum());
            } catch (Throwable t) {
                logger.error("[updateSlotState]update slot table:{} error.", update, t);
                return false;
            } finally {
                writeLock.unlock();
            }
            return true;
        }

        Set<Integer> slotIds() {
            readLock.lock();
            try {
                return slotStates.keySet();
            } finally {
                readLock.unlock();
            }
        }

        boolean addSlotState(Set<Integer> slotIds) {
            if (CollectionUtils.isEmpty(slotIds)) {
                return true;
            }
            writeLock.lock();
            try {
                for (Integer slotId : slotIds) {
                    Slot slot = slots.getSlot(slotId);
                    if (slot == null) {
                        logger.error("[addSlotState]add dataCenter={}, slotId={} fail, " + "for slotId not exist in Slots={}", dataCenter, slotId, slots);
                        continue;
                    }
                    listenAdd(dataCenter, slotId);
                    slotStates.computeIfAbsent(slotId, k -> {
                        logger.info("[addSlotState]add dataCenter={}, slot={}", dataCenter, slot);
                        return new RemoteSlotStates(dataCenter, slot);
                    });
                }
            } catch (Throwable t) {
                logger.error("[addSlotState]add dataCenter={}, slotIds={} error.", dataCenter, slotIds, t);
                return false;
            } finally {
                writeLock.unlock();
            }
            return true;
        }

        boolean removeSlotState(Set<Integer> slotIds) {
            if (CollectionUtils.isEmpty(slotIds)) {
                return true;
            }
            writeLock.lock();
            try {
                for (Integer slotId : slotIds) {
                    slotStates.remove(slotId);
                    // first remove the slot for GetData Access check, then clean the data
                    listenRemove(dataCenter, slotId);
                    MultiClusterSlotMetrics.observeRemoteLeaderSyncingFinish(dataCenter, slotId);
                    logger.info("[removeSlotState]dataCenter={} remove slot, slotId={}", dataCenter, slotId);
                }
            } catch (Throwable t) {
                logger.error("[removeSlotState]remove dataCenter={}, slotIds={} error.", dataCenter, slotIds, t);
                return false;
            } finally {
                writeLock.unlock();
            }
            return true;
        }

        long getSlotsEpoch() {
            readLock.lock();
            try {
                return slots.getEpoch();
            } finally {
                readLock.unlock();
            }
        }

        List<BaseSlotStatus> getSlotStatuses() {
            List<BaseSlotStatus> slotStatuses = new ArrayList<>(slotStates.size());
            updateLock.readLock().lock();
            try {
                for (Map.Entry<Integer, RemoteSlotStates> entry : slotStates.entrySet()) {
                    int slotId = entry.getKey();
                    RemoteSlotStates slotState = entry.getValue();
                    LeaderSlotStatus status = new LeaderSlotStatus(slotId, slotState.slot.getLeaderEpoch(), slotState.slot.getLeaderDataStoreNodeId(), slotState.synced ? BaseSlotStatus.LeaderStatus.healthy : BaseSlotStatus.LeaderStatus.unhealthy);
                    slotStatuses.add(status);
                }

                Collections.sort(slotStatuses, Comparator.comparingInt(s -> s.getSlotId()));
                return slotStatuses;
            } finally {
                updateLock.readLock().unlock();
            }
        }

        RemoteSlotStates getSlotStates(int slotId) {
            readLock.lock();
            try {
                return slotStates.get(slotId);
            } finally {
                readLock.unlock();
            }
        }

        Tuple<Slots, RemoteSlotStates> get(int slotId) {
            Slots table;
            RemoteSlotStates state;
            readLock.lock();
            try {
                table = slots;
                state = slotStates.get(slotId);
            } finally {
                readLock.unlock();
            }
            return new Tuple<>(table, state);
        }

        private void listenAdd(String dataCenter, int slotId) {
            slotChangeListenerManager.remoteListeners().forEach(listener -> listener.onSlotAdd(dataCenter, slotId, SlotRole.leader));
        }

        private void listenRemove(String dataCenter, int slotId) {
            slotChangeListenerManager.remoteListeners().forEach(listener -> listener.onSlotRemove(dataCenter, slotId, SlotRole.leader));
        }
    }

    final class SyncDataIdTask extends SyncLeaderTask {

        private final Set<String> syncing;

        public SyncDataIdTask(String localDataCenter, String syncDataCenter, long slotsEpoch, Slot slot, SlotDiffSyncer syncer, Set<String> syncing, ClientSideExchanger clientSideExchanger, SyncContinues continues) {
            super(localDataCenter, syncDataCenter, slotsEpoch, slot, syncer, clientSideExchanger, continues);
            this.syncing = syncing;
        }


    }

    final class RemoteSyncingWatchDog extends WakeUpLoopExecuteTask {

        @Override
        public void _execute_() {
            try {
                doUpdating();
                watchLocalSegSlot();
                doSyncRemoteLeader();

            } catch (Throwable t) {
                logger.error("[remoteSyncWatch]failed to do sync watching.", t);
            }
        }

        @Override
        public int getWaitingMillis() {
            return 200;
        }
    }
}
